package com.lubase.metadata.service.impl;

import com.lubase.core.service.flowform.WorkFlowAdapterService;
import com.lubase.metadata.model.ELakeStatus;
import com.lubase.metadata.service.ApplyLakeService;
import com.lubase.model.DbEntity;
import com.lubase.orm.QueryOption;
import com.lubase.orm.TableFilter;
import com.lubase.orm.exception.WarnCommonException;
import com.lubase.model.DbCollection;
import com.lubase.orm.operate.EOperateMode;
import com.lubase.orm.service.AppHolderService;
import com.lubase.orm.service.DataAccess;
import com.lubase.orm.util.TableFilterWrapper;
import com.lubase.orm.util.TypeConverterUtils;
import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Service
public class ApplyLakeServiceImpl implements ApplyLakeService {
    @Autowired
    DataAccess dataAccess;
    @Autowired
    WorkFlowAdapterService workFlowAdapterService;
    @Autowired
    AppHolderService appHolderService;
    /**
     * 入湖申请流程业务ID
     */
    private String applyLakeWfId = "1481919486398304256";

    @SneakyThrows
    @Override
    public Integer submitApplyLake(String lakeId) {
        // 检查参数
        if (lakeId == null || lakeId.isEmpty()) {
            throw new WarnCommonException("submitApplyLake: lakeId is empty");
        }
        QueryOption queryOption = new QueryOption("mt_lake_apply");
        queryOption.setTableFilter(new TableFilter("id", lakeId));
        DbCollection collLakeApply = dataAccess.queryAllData(queryOption);
        if (collLakeApply.getData().isEmpty()) {
            throw new WarnCommonException("提交失败：申请Id不存在");
        }

        queryOption = new QueryOption("mt_lake_detail");
        queryOption.setTableFilter(new TableFilter("mt_lake_apply_id", lakeId));
        DbCollection collLakeDetail = dataAccess.queryAllData(queryOption);
        if (collLakeDetail.getData().isEmpty()) {
            throw new WarnCommonException("提交失败：请选择入湖的表");
        }

        queryOption = new QueryOption("mt_table");
        TableFilterWrapper tableFilterWrapper = TableFilterWrapper.or();
        for (DbEntity entity : collLakeDetail.getData()) {
            tableFilterWrapper.eq("id", entity.get("mt_table_id").toString());
        }
        queryOption.setTableFilter(tableFilterWrapper.build());
        queryOption.setFixField("zh_percentage,table_code,merge_name,lake_status");
        DbCollection collTable = dataAccess.queryAllData(queryOption);
        for (DbEntity table : collTable.getData()) {
            if (TypeConverterUtils.object2Integer(table.get("zh_percentage"), 0) != 100) {
                throw new WarnCommonException(String.format("提交失败：入湖表 %s 中文注释未达%s，请检查", table.get("table_code"), "100%"));
            }
            if (TypeConverterUtils.object2Integer(table.get("lake_status"), 0) != 0) {
                throw new WarnCommonException(String.format("提交失败：入湖表 %s 已入湖或入湖中，请检查", table.get("table_code")));
            }
        }

        List<String> databaseIdList = new ArrayList<>();
        List<String> systemIdList = new ArrayList<>();
        List<String> dbaList = new ArrayList<>();
        List<String> systemOwnerList = new ArrayList<>();
        collLakeDetail.getData().forEach(dbEntity -> {
            databaseIdList.add(dbEntity.get("ms_database_id").toString());
        });

        queryOption = new QueryOption("ms_database_info");
        queryOption.setTableFilter(new TableFilter("ms_database_id", databaseIdList.stream().distinct().collect(Collectors.joining(",")), EOperateMode.In));
        DbCollection collDatabaseInfo = dataAccess.queryAllData(queryOption);
        if (collDatabaseInfo.getData().isEmpty()) {
            throw new WarnCommonException("提交失败：入湖表数据源配置错误");
        }

        for (DbEntity entity : collLakeDetail.getData()) {
            // 检查入湖表的数据源配置是否正确
            String databaseId = entity.get("ms_database_id").toString();
            String lakeEnv = entity.get("lake_env").toString();
            for (String env : lakeEnv.split(",")) {
                if (collDatabaseInfo.getData().stream().noneMatch(dbEntity -> dbEntity.get("ms_database_id").toString().equals(databaseId) && dbEntity.get("env_info").toString().equals(env))) {
                    throw new WarnCommonException(String.format("提交失败：入湖表 %s 选择的环境信息在数据源配置中缺失，请检查", entity.getRefData().get("mt_table_idNAME")));
                }
            }
        }

        queryOption = new QueryOption("ms_database");
        queryOption.setTableFilter(new TableFilter("id", databaseIdList.stream().distinct().collect(Collectors.joining(",")), EOperateMode.In));
        DbCollection collDatabase = dataAccess.queryAllData(queryOption);
        collDatabase.getData().forEach(dbEntity -> {
            systemIdList.add(dbEntity.get("ms_system_list_id").toString());
            dbaList.add(dbEntity.get("db_manager").toString());
        });
        queryOption = new QueryOption("ms_system_list");
        queryOption.setTableFilter(new TableFilter("id", systemIdList.stream().distinct().collect(Collectors.joining(",")), EOperateMode.In));
        DbCollection collSystemList = dataAccess.queryAllData(queryOption);
        collSystemList.getData().forEach(dbEntity -> {
            systemOwnerList.add(dbEntity.get("system_manager").toString());
        });

        // 更新系统负责人和数据库DBA、提交流程
        collLakeApply.getData().get(0).put("db_owner", dbaList.stream().distinct().collect(Collectors.joining(",")));
        collLakeApply.getData().get(0).put("system_owner", systemOwnerList.stream().distinct().collect(Collectors.joining(",")));
        dataAccess.update(collLakeApply);

        workFlowAdapterService.startWfByApi(applyLakeWfId, lakeId, appHolderService.getUser().getId().toString());

        // 更新入湖标记
        collTable.getData().forEach(dbEntity -> {
            dbEntity.put("lake_status", ELakeStatus.Laking.getType());// 入湖状态1 入湖
        });
        collTable.setEnableTableTrigger(false);
        return dataAccess.update(collTable);
    }

    @SneakyThrows
    @Override
    public Integer applyLakeComplete(String lakeId) {
        // 检查参数
        if (lakeId == null || lakeId.isEmpty()) {
            throw new WarnCommonException("submitApplyLake: lakeId is empty");
        }
        QueryOption queryOption = new QueryOption("mt_lake_apply");
        queryOption.setTableFilter(new TableFilter("id", lakeId));
        DbCollection collLakeApply = dataAccess.queryAllData(queryOption);
        if (collLakeApply.getData().isEmpty()) {
            throw new WarnCommonException("提交失败：申请Id不存在");
        }

        queryOption = new QueryOption("mt_lake_detail");
        queryOption.setFixField("mt_table_id,lake_table_code");
        queryOption.setTableFilter(new TableFilter("mt_lake_apply_id", lakeId));
        DbCollection collLakeDetail = dataAccess.queryAllData(queryOption);
        if (collLakeDetail.getData().isEmpty()) {
            throw new WarnCommonException("提交失败：请选择入湖的表");
        }

        queryOption = new QueryOption("mt_table");
        TableFilterWrapper tableFilterWrapper = TableFilterWrapper.or();
        for (DbEntity entity : collLakeDetail.getData()) {
            tableFilterWrapper.eq("id", entity.get("mt_table_id").toString());
        }
        queryOption.setTableFilter(tableFilterWrapper.build());
        queryOption.setFixField("lake_status");
        DbCollection collTable = dataAccess.queryAllData(queryOption);

        // 更新入湖标记
        collTable.getData().forEach(dbEntity -> {
            dbEntity.put("lake_status", ELakeStatus.Laked.getType());// 入湖状态1 入湖
            DbEntity lakeEntity = collLakeDetail.getData().stream().filter(entity -> entity.get("mt_table_id").toString().equals(dbEntity.get("id").toString())).findFirst().orElse(null);
            if (lakeEntity != null) {
                dbEntity.put("lake_table_code", lakeEntity.get("lake_table_code"));
            }
        });
        collTable.setEnableTableTrigger(false);
        return dataAccess.update(collTable);
    }
}
